package com.intct.func;

import com.intct.hbase.bean.LoucsBean;
import com.intct.util.JedisUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

/**
 * @author gufg
 * @since 2025-08-21 14:35
 */
public class AdsLocusRedisSink extends RichSinkFunction<LoucsBean> {

    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = JedisUtil.getJedis();
    }

    @Override
    public void invoke(LoucsBean loucsBean, Context context) throws Exception {
        String driverId = loucsBean.getDriverId();
        // 如果订单结束
        if ("end".equals(driverId)) {
            // 删除实时订单集合
            jedis.srem("retime", loucsBean.getOrderId());
            // 增加到历史订单集合
            jedis.sadd("history", loucsBean.getOrderId());
        } else {
            // 增加到实时订单集合
            jedis.sadd("retime", loucsBean.getOrderId());
        }
        // 使用Redis的有序集合
        jedis.zadd(loucsBean.getOrderId(), loucsBean.getTime(), loucsBean.getLat() + ":" + loucsBean.getLat());
    }

    @Override
    public void close() throws Exception {
        JedisUtil.close(jedis);
    }
}
